package edu.kit.ksri.db.mturk.qualifications;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.StringTokenizer;

import org.apache.log4j.Logger;

import com.amazonaws.mturk.requester.QualificationRequest;
import com.amazonaws.mturk.service.axis.RequesterService;
import com.amazonaws.mturk.util.PropertiesClientConfig;

import edu.kit.ksri.db.mturk.connectionhandlers.WorkerManager;
import edu.kit.ksri.db.mturk.plugins.correctness.answers.BasicAssignmentAnswer;
import edu.kit.ksri.db.mturk.plugins.display.BrowserDisplay;
import edu.kit.ksri.db.mturk.var.Icd;
import edu.kit.ksri.db.mturk.var.StaticInformation;
import edu.kit.ksri.db.mturk.workerpools.AbstractWorker;
import edu.kit.ksri.db.mturk.workerpools.ICDWorker;
import edu.kit.ksri.db.mturk.workerpools.PoolManager;
import edu.kit.ksri.db.mturk.workerpools.StandardWorker;

/**
 * Copyright (c) 2010-2011 Karlsruhe Institute of Technology (KIT), Germany
 *
 * Permission is hereby granted, free of charge, to any person obtaining a
 * copy of this software and associated documentation files (the "Software"),
 * to deal in the Software without restriction, including without limitation
 * the rights to use, copy, modify, merge, publish, distribute, sublicense,
 * and/or sell copies of the Software, and to permit persons to whom the
 * Software is furnished to do so, subject to the following conditions:
 * 
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 * DEALINGS IN THE SOFTWARE.
 * 
 * --------------------------------------------------------------------------
 * 
 * Software: CSP/WMV tool for dynamic quality management of microtasks
 * http://www.ksri.kit.edu/Upload/Users/PeopleClouds/HCOMP2011_An_extendable_toolkit_for_managing_quality_of_human-based_eServices.pdf
 * 
 * Organization: Karlsruhe Institute of Technology (KIT), Germany
 * http://www.kit.edu
 * 
 * Project: PeopleClouds project of the Karlsruhe Service Research Institute(KSRI)
 * http://www.ksri.kit.edu/Default.aspx?PageId=818
 * 
 * Contributing authors: David Bermbach, Sandra Rath, Pascal Wichmann, Robert Kern
 */
public class QualificationManager implements Runnable {

	private static final Logger basiclog = Logger.getLogger("basic."
			+ QualificationManager.class);
	
	private static final Logger qualilog = Logger.getLogger("quali."
			+ QualificationManager.class);

	/** if set to false the {@link QualificationManager} will terminate */
	private static boolean running = true;

	/** will be set to true once the run method terminates */
	private static boolean terminated = false;

	public void run() {
		

		RequesterService rs = new RequesterService(new PropertiesClientConfig(
				StaticInformation.mturkFile));
		while (running) {
			try {
				Thread.sleep(StaticInformation.millisBetweenMTurkPolling);
			} catch (InterruptedException e) {
			}
			QualificationRequest[] requests = rs
					.getQualificationRequests(StaticInformation.scenario
							.getQualificationTypeID());
			if (requests == null)
				continue;

			basiclog.info("Got " + requests.length + " qualification requests.");
			//BEGIN ROB
			//BrowserDisplay.update(requests.length);
			// END ROB	
			for (QualificationRequest q : requests) {
				String workerid = q.getSubjectId();
				// parse answer
				String answer = q.getAnswer();
				if (answer == null) {
					rs.rejectQualificationRequest(
							q.getQualificationRequestId(),
							"There was an error processing your"
									+ " qualification request. Please,"
									+ " try again later.");
					continue;
				}

// -------- ICD scenario ---------
				if (StaticInformation.scenario instanceof Icd) {
					
					// read gold standard
					ArrayList<LinkedList<String>> goldstandard = new ArrayList<LinkedList<String>>();
					try {
						//read the qualification test answers file
						BufferedReader in = new BufferedReader(new InputStreamReader(
								new FileInputStream(
										StaticInformation.qualificationTestAnswersFile)));
						String line = in.readLine();
						
						//counter for atomic answers (code + flags)
						int atomicAnswers = 0;
						
						//in case more than one diagnose should be tested
						LinkedList<String> compositeAnswer = new LinkedList<String>();
						
						while (line != null) {
							StringTokenizer st = new StringTokenizer(line, ",");
							//So far we don't need the key, but maybe later in time we want to use it
							//The value is one atomic answer (code + flags)
							st.nextToken();//key
							String value = st.nextToken();
							//Each composite answer consists of several atomic answers
							compositeAnswer.add(value);
							line = in.readLine();
							atomicAnswers++;
							
							if(atomicAnswers % Icd.maxNumberOfLabelsQualiTest == 0){
								//in case more than one diagnose should be tested
								goldstandard.add(compositeAnswer);
								compositeAnswer = new LinkedList<String>();
							}
						}
					} catch (Exception e) {
						e.printStackTrace();
						basiclog.error("Could not read file with"
								+ " gold standard. Terminating" + " QualificationManager.");
						basiclog.info("Terminated.");
						terminated = true;
						return;
					}
					
					
					//read worker answer
					int truePositives = 0, trueNegatives = 0, falsePositives = 0, falseNegatives = 0;
					HashMap<String, String> answers = BasicAssignmentAnswer
							.parseAnswer(answer);
					
					basiclog.debug("Current parsed answers are " + answers.toString());
					
					//loops states how many composite answers we have, in case more than one diagnose should be tested
					int loops = (int) (answers.size() / (2*Icd.maxNumberOfLabelsQualiTest));
					
					//This ArrayList stores 1 or more composite answers
					ArrayList<LinkedList<String>> compositeAnswers = new ArrayList<LinkedList<String>>();
					
					//Do this as often as we have diagnoses (usually just 1)
					for(int i = 0; i < loops; i++){
						//This LinkedList stores the atomic answers for one diagnose (so this is one composite answer)
						LinkedList<String> compositeAnswer = new LinkedList<String>();
						
						//Do this for every label (equals atomic answer) we have
						for(int j = 1; j <= Icd.maxNumberOfLabelsQualiTest; j++){
							
							//Get the code
							String code = answers.get("code"+String.valueOf(j));
							
							if(!code.equals("")){
								//Code is not empty, so check the flags
								String flags = answers.get("code"+String.valueOf(j)+"flags");
								
								if(!flags.equals("")){
									//Flags are not empty, so include the code in the label
									String label = code;
									
									//Now concatenate the flags to the label
									for(int k = 1; k <= Icd.flagCount; k++){
										if (flags.contains("flag"+k)) {
											//Flag is contained, so the flag has been set
											label = label + " true";
										}
										else{
											//Flag ist not contained, so the flag has not been set
											label = label + " false";
										}
									}
									
									//Add the label for the collection of labels (collection is the composite answer)
									compositeAnswer.add(label);
									basiclog.info("Label " + j + " is " + label);
								}
								else{
									//Flags are empty, so include the code in the label and set the flags to false
									String label = code+" false false false false";
									compositeAnswer.add(label);
									basiclog.info("Label " + j + " is " + label);
								}
							}
							else{
								//Code is empty, so set the flags to false
								String label = "false false false false";
								compositeAnswer.add(label);
								basiclog.info("Label " + j + " is " + label);
							}
						}
						
						//We are done with the composite answer, now add it to the Array List that contains all composite answers
						compositeAnswers.add(compositeAnswer);
					}
					
					
					// Compare goldstandard and worker answer
					for(int i = 0; i < goldstandard.size(); i++){
						LinkedList<String> goldCompositeAnswer = goldstandard.get(i);
						LinkedList<String> workerCompositeAnswer = compositeAnswers.get(i);
						
						//Each composite answer consists of several atomic answers
						//Take the first gold standard atomic answer
						for(String correctAtomicAnswer : goldCompositeAnswer){
//							in case a concurrent exception occurs: take for loop
//							for(int j=0; j<goldCompositeAnswer.size(); j++){
//							String correctAtomicAnswer = goldCompositeAnswer.get(j);
							
							//remove flags to check if code is empty
							String emptyCheck = correctAtomicAnswer.replace("true", "");
							emptyCheck = emptyCheck.replace("false", "");
							emptyCheck = emptyCheck.trim();
							
							//Does the worker answer contain the correct atomic answer (label)?
							if(workerCompositeAnswer.contains(correctAtomicAnswer)){
								//Yes, worker answer contains the correct label
								if(emptyCheck.equals("")){
									//code is empty and the label is contained in the worker answer, so it is a true negative
									//trueNegatives (TN) are empty/default-value labels in both gold standard and assignment answer
									trueNegatives++;
								}
								else{
									//code is not empty and the label is contained in the worker answer, so it is a true positive
									//truePositives (TP) are those (non-empty/non-default) labels in both gold standard and assignment answer
									truePositives++;
								}
								
								//Remove the answer from the worker answers to avoid counting the same answer twice
								workerCompositeAnswer.remove(correctAtomicAnswer);								
							}
							else{
								//No, worker answer does not contain the correct label
								if(!emptyCheck.equals("")){
									//code is not empty and the label is not contained in the worker answer
									//falseNegatives (FN) are those (non-empty/non-default) labels only in the gold standard and NOT in the assignment answer.
									falseNegatives++;
								}
							}
						}
						
						//Now we check for false positives
						//This is done by looping over the remaining worker answers that are not in the gold standard...
						//...and counting them in case they are not empty
						for(int j = 0; j < workerCompositeAnswer.size(); j++){
							String workerAnswer = workerCompositeAnswer.get(j);
							
							//remove flags to check if code is empty
							String emptyCheck = workerAnswer.replace("true", "");
							emptyCheck = emptyCheck.replace("false", "");
							emptyCheck = emptyCheck.trim();

							if(!emptyCheck.equals("")){
								//code is not empty and the label is not contained in the gold standard
								//falsePositives (FP) are those (non-empty/non-default) labels only in the assignment answer and NOT in the gold standard
								falsePositives++;
							}
						}
					}
					
					//Now we calculate the sensitivity and the specificity by using the TP, FP, TN, FN
					double sensitivity = (double) truePositives/(truePositives + falseNegatives);
					double specificity = (double) trueNegatives/(falsePositives + trueNegatives);

					basiclog.info("Test result for worker " + workerid + ": "
							+ truePositives + " TP, " + trueNegatives + " TN, " 
							+ falsePositives + " FP, " + falseNegatives + " FN. "
							+ "The specificity is " + specificity + " and the sensitivity is "+ sensitivity + ".");
					
					//Create new ICD worker
					AbstractWorker w = new ICDWorker(workerid, goldstandard.size(), specificity, sensitivity);

					//Get the score for this worker
					Integer score = WorkerManager.getInstance()
							.getQualificationValue(workerid);
					//Check if score is set and higher than the pendingPoolQualificationValue
					if (score != null
							&& score > StaticInformation.pendingPoolQualificationValue) {
						basiclog.info("Worker " + workerid
								+ " is already part of the active"
								+ " pool and will not be moved"
								+ " to the pending pool.");
						qualilog.info("ACTIVE," + workerid + "," + goldstandard.size() + "," + specificity + "," + sensitivity);
						PoolManager.getInstance().getPoolForName("ACTIVE")
								.addWorker(w);
						rs.grantQualification(q.getQualificationRequestId(),
								StaticInformation.activePoolQualificationValue);
						continue;
					}
					//Score is null, so we check if the worker reached a certain sensitivity and specificity
					if (sensitivity > 0.5 && specificity > 0.5) {
						//Worker has reached a certain sensitivity and specificity, so add it to the active pool
						basiclog.info("Worker " + workerid
								+ " is added to the active pool.");
						qualilog.info("ACTIVE," + workerid + "," + goldstandard.size() + "," + specificity + "," + sensitivity);
						PoolManager.getInstance().getPoolForName("ACTIVE")
								.addWorker(w);
						rs.grantQualification(q.getQualificationRequestId(),
								StaticInformation.activePoolQualificationValue);
						continue;
					}
					
					//Worker did not reach a certain sensitivity and specificity, so add it to the pending pool
					rs.grantQualification(q.getQualificationRequestId(),
							StaticInformation.pendingPoolQualificationValue);
					basiclog.info("Worker is added to the pending pool.");
					qualilog.info("PENDING," + workerid + "," + goldstandard.size() + "," + specificity + "," + sensitivity);
					PoolManager.getInstance().getPoolForName("PENDING")
							.addWorker(w);
				}
				else{
//--------------Standard Worker--------------------				
					// read gold standard
					HashMap<String, String> goldstandard = new HashMap<String, String>();
					try {
						BufferedReader in = new BufferedReader(new InputStreamReader(
								new FileInputStream(
										StaticInformation.qualificationTestAnswersFile)));
						String line = in.readLine();
						while (line != null) {
							StringTokenizer st = new StringTokenizer(line, ",");
							String key = st.nextToken(), value = st.nextToken();
							goldstandard.put(key, value);
							line = in.readLine();
						}
					} catch (Exception e) {
						e.printStackTrace();
						basiclog.error("Could not read file with"
								+ " gold standard. Terminating" + " QualificationManager.");
						basiclog.info("Terminated.");
						terminated = true;
						return;
					}
					
					// read worker answer
					int correct = 0, wrong = 0;
					HashMap<String, String> answers = BasicAssignmentAnswer
							.parseAnswer(answer);
					
					// compare to gold standard
					for (String key : goldstandard.keySet()) {
						try {
							if (goldstandard.get(key).trim()
									.equals(answers.get(key).trim()))
								correct++;
							else
								wrong++;
						} catch (NullPointerException e) {
							wrong++;
						}
					}
					
					// ROB 
					// Keep initial worker error rates above zero!
					if (correct == 10) {
						correct--;
						wrong++;
					}
					// ROB

					
					StandardWorker w = new StandardWorker(workerid, goldstandard.size(), correct,
								wrong);
	
					Integer score = WorkerManager.getInstance()
							.getQualificationValue(workerid);
					if (score != null
							&& score > StaticInformation.pendingPoolQualificationValue) {
						basiclog.info("Test result for worker " + workerid + ": "
								+ correct + " correct and " + wrong + " wrong (of "
								+ goldstandard.size() + "). Worker is already part of the " +
								"active pool and will not be moved to the pending pool.");
						qualilog.info("ACTIVE," + workerid + "," + goldstandard.size() + "," + correct + "," + wrong);
						PoolManager.getInstance().getPoolForName("ACTIVE")
								.addWorker(w);
						rs.grantQualification(q.getQualificationRequestId(),
								StaticInformation.activePoolQualificationValue);
						continue;
					}
					if (wrong == 1) {
						PoolManager.getInstance().getPoolForName("ACTIVE")
								.addWorker(w);
						basiclog.info("Test result for worker " + workerid + ": "
								+ correct + " correct and " + wrong + " wrong (of "
								+ goldstandard.size() + "). Worker is added to the active pool.");
						qualilog.info("ACTIVE," + workerid + "," + goldstandard.size() + "," + correct + "," + wrong);
						rs.grantQualification(q.getQualificationRequestId(),
								StaticInformation.activePoolQualificationValue);
						continue;
					}
					rs.grantQualification(q.getQualificationRequestId(),
							StaticInformation.pendingPoolQualificationValue);
					basiclog.info("Test result for worker " + workerid + ": "
							+ correct + " correct and " + wrong + " wrong (of "
							+ goldstandard.size() + "). Worker is added to the pending pool.");
					qualilog.info("PENDING," + workerid + "," + goldstandard.size() + "," + correct + "," + wrong);
					PoolManager.getInstance().getPoolForName("PENDING")
							.addWorker(w);
				}
			}
		}
		basiclog.info("Terminated.");
		terminated = true;
	}

	/**
	 * @return the running
	 */
	public static boolean isRunning() {
		return running;
	}

	/**
	 * @param running
	 *            the running to set
	 */
	public static void setRunning(boolean running) {
		QualificationManager.running = running;
	}

	/**
	 * @return the terminated
	 */
	public static boolean isTerminated() {
		return terminated;
	}

	/**
	 * @param terminated
	 *            the terminated to set
	 */
	public static void setTerminated(boolean terminated) {
		QualificationManager.terminated = terminated;
	}

}
